package com.szwistar.common.comm.jsonrpc.mina;

import com.shephertz.appwarp.websocket.binary.WebSocketCodecFactory;
import com.szwistar.common.comm.InetSocketAddressEx;
import com.szwistar.common.comm.ServerBase;
import com.szwistar.common.comm.jsonrpc.*;
import com.szwistar.common.datastruct.ObjectMethod;
import com.szwistar.common.datastruct.Reflector;
import com.szwistar.common.datastruct.Utils;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.LineDelimiter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.URI;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * JsonRpc 的 Mina 实现
 * @author 张勇
 */
public class MinaJsonRpcServer extends ServerBase<MinaJsonRpcCommEndpoint> {
	protected static final Logger LOG = LoggerFactory.getLogger(MinaJsonRpcServer.class);

	/**
	 * 创建 JsonRpc@WebSocket 服务器
	 * @param address WebSocket 服务器地址，如："*:8504"
	 */
	private static MinaJsonRpcServer createWebsocketDeamon(String name, String address) throws Exception {
		// 创建服务器
		InetSocketAddress addr = InetSocketAddressEx.fromString(address);
		MinaJsonRpcServer server = new MinaJsonRpcServer(name, addr, true);

		// 设置编解码器
		server.getAcceptor().getFilterChain().addFirst("webSocketCodec", new ProtocolCodecFilter(new WebSocketCodecFactory()));
		return server;
	}

	/**
	 * 创建 JsonRpc@TCP/UDP 服务器
	 * <p>数据以 JSON 格式字符串传输，以 \n 做为结束符；中文使用 GBK 编码。
	 * @param address 服务器监听地址和端口，如："*:9505"
	 * @param tcp true: 使用 TCP，false: 使用 UDP
	 */
	private static MinaJsonRpcServer createSocketDeamon(String name, String address, boolean tcp) throws Exception {
		// 创建服务器
		InetSocketAddress addr = InetSocketAddressEx.fromString(address);
		MinaJsonRpcServer server = new MinaJsonRpcServer(name, addr, tcp);

		// 设置编解码器
		String encode = "GBK";
		String decodingDelimiter = LineDelimiter.UNIX.getValue();
		String encodingDelimiter = LineDelimiter.UNIX.getValue();

		TextLineCodecFactory codecFactory = new TextLineCodecFactory(Charset.forName(encode), encodingDelimiter, decodingDelimiter);
		codecFactory.setDecoderMaxLineLength(256*1024);
		codecFactory.setEncoderMaxLineLength(256*1024);

		server.getAcceptor().getFilterChain().addLast("luaRpcCodec", new ProtocolCodecFilter(codecFactory));
		return server;
	}

	/**
	 * 根据 address 标识，确定服务器的实现方式和监听地址，并创建 MinaJsonRpcServer
 	 */
	public static MinaJsonRpcServer create(String name, String address) throws Exception {
		MinaJsonRpcServer server;
		URI uri = new URI(address);

		if(uri.getScheme().equals("ws") || uri.getScheme().equals("wss")) {
			server = createWebsocketDeamon(name, uri.getAuthority());
		} else if(uri.getScheme().equals("tcp")) {
			server = createSocketDeamon(name, uri.getAuthority(), true);
		} else if(uri.getScheme().equals("udp")) {
			server = createSocketDeamon(name, uri.getAuthority(), false);
		} else {
			throw new Exception("不支持的服务地址：" + address);
		}

		return server;
	}

	// 本端自己的监听 Topic，对于 P2p 类连接不起作用，对 MQTT 连接有效
	protected String myLocalTopic;
	// 提供的 rpc 服务函数表
	protected Map<String, ObjectMethod> methodsMap = new HashMap<String, ObjectMethod>();
	// 提供的 notify 服务函数表
	protected Map<String, ObjectMethod> eventsMap = new HashMap<String, ObjectMethod>();

	// 获取注册的 rpc 服务函数表
	protected Map<String, ObjectMethod> getMethodsMap() { return methodsMap; }
	// 获取注册的 notify 服务函数表
	protected Map<String, ObjectMethod> getEventsMap() { return eventsMap; }

	// 不允许直接创建 MinaJsonRpcServer 对象，应使用 create() 方法创建
	private MinaJsonRpcServer(String name, InetSocketAddress address, boolean tcp) {
		super(name, address, MinaJsonRpcCommEndpoint.class, null, tcp);
		clientContext = this;
	}

	/**
	 * 注册服务：在此处注册的服务适用于所有的子连接
	 */
	public void registerService(JsonRpcService service) {
		List<Method> methods;

		// 登记所有 RPC 服务函数
		methods = Reflector.getMethods(service.getClass(), null, null, JsonRpcMessageReturn.class, JsonRpcMessageCall.class);
		for(Method method: methods) {
			// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
			String name = service.getName() + "." + method.getName();
			//LOG.debug("登记 RPC 函数：{} => {}.{}()", name, service.getClass().getName(), method.getName());
			methodsMap.put(name, new ObjectMethod(service, method));
		}

		// 登记所有事件处理函数
		// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
		methods = Reflector.getMethods(service.getClass(), null, null, void.class, JsonRpcMessageNotify.class);
		for(Method method: methods) {
			// 服务中的服务函数调用名为："服务名称.函数名"，如： "xiaodu.discover"
			String name = service.getName() + "." + method.getName();
			//LOG.debug("登记事件处理函数：{} => {}.{}()", name, service.getClass().getName(), method.getName());
			eventsMap.put(name, new ObjectMethod(service, method));
		}
	}

	/**
	 * 注册服务集合：在此处注册的服务适用于所有的子连接
	 */
	public void registerServices(Collection<JsonRpcService> services) {
		for(JsonRpcService service: services) {
			registerService(service);
		}
	}

	/**
	 * 获得指定来源的客户端
 	 */
	public MinaJsonRpcCommEndpoint getClient(String from) {
		if(Utils.isEmpty(from)) { return null; }
		for(MinaJsonRpcCommEndpoint client: getClients().values()) {
			if(from.equals(client.getFrom())) { return client; }
		}
		return null;
	}

	/**
	 * 尝试向所有连接的客户端广播通知（如果对端未订阅此主题，则会跳过）
	 * @see com.szwistar.common.comm.jsonrpc.JsonRpcEndpointAbstract#notify(String, String, String, Object)
	 */
	public void notify(String targetTopic, String token, String name, Object params) {
		// 构造 notify 消息
		JsonRpcMessage msg;
		try {
			msg = new JsonRpcMessageNotify(myLocalTopic, token, name, JsonRpcEndpointAbstract.packParam(params));
		} catch (Exception e) {
			LOG.error("参数编码失败！{}", e.getMessage());
			return;
		}

		// 遍历所有连接的客户端
		for(MinaJsonRpcCommEndpoint client: getClients().values()) {
			// 不直接调用子连接发送 notify，原因是避免重复创建 msg，提升效率
			//client.notify(targetTopic, token, name, params);
			// 如果对端未订阅此主题，则会跳过
			if(!client.isPeerSubscribed(targetTopic)) continue;
			// 发送通知消息
			LOGCOMM.debug("向 '{}' 发送主题为 '{}' 的通知：{}", client, targetTopic, msg);
			client.send(targetTopic, msg, 0, 1);
		}
	}
}
